package com.shujia.spark.mllib

import org.apache.spark.ml.feature.LabeledPoint
import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.ml.regression.{LinearRegression, LinearRegressionModel}
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Demo1Point {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("point")
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._

    /**
     * 1、读取原始数据
     */
    //读取数据
    val pointsDF: DataFrame = spark.read
      .format("csv")
      .option("sep", ",")
      .schema("label DOUBLE,features DOUBLE")
      .load("data/points.txt")

    /**
     * 2、特征工程：将原始的数据转换成算法可以识别的LabeledPoint格式的数据
     *
     */
    //将特征转换成特征向量
    val labeledPointDS: Dataset[LabeledPoint] = pointsDF.map {
      case Row(label: Double, features: Double) =>
        //构建标记数据
        LabeledPoint(label, Vectors.dense(features))
    }
    //转换成DF
    val labeledPointDF: DataFrame = labeledPointDS.toDF()


    /**
     * 3、选择算法
     */
    //线性回归算法（y=w1*x1+w2*x2+....+wn*xn + w）
    val linearRegression = new LinearRegression()

    /**
     * 4、将训练数据带入算法训练模型
     * 模型：公式 + k + b
     */
    //内部会使用spark进行循环迭代计算
    val model: LinearRegressionModel = linearRegression.fit(labeledPointDF)

    //截距
    println(model.intercept)
    //斜率（权重）
    println(model.coefficients)
    // y = 0.2893 * x -0.1670


    //对新的x做预测
    val y: Double = model.predict(Vectors.dense(Array(10.0)))
    println(y)

  }
}
